package com.sdhs.paas.flux.app;

import java.io.IOException;
import java.time.Duration;

import reactor.core.Exceptions;
import reactor.core.publisher.Flux;

public class TestFluxDemo1 {
	public static void main(String[] args) throws InterruptedException {
		Flux.interval(Duration.ofMillis(250)).map(input -> {
			if (input < 3)
				return "tick " + input;
			throw new RuntimeException("boom");
		}).elapsed().retry(1).subscribe(System.out::println, System.err::println);

		Thread.sleep(2100);
		System.out.println("--------------retrywhen------------------------------");
		Flux<String> fluxRetryWhen = Flux.<String> error(new IllegalArgumentException()).doOnError(System.out::println)
				.retryWhen(companion -> companion.take(3));
		fluxRetryWhen.subscribe(i -> System.out.println(i));
		System.out.println("--------------retrywhen--------------------------------");
		Flux.just("foo").map(s -> {
			throw new IllegalArgumentException(s);
		}).subscribe(v -> System.out.println("GOT VALUE"), e -> System.out.println("ERROR: " + e));
		System.out.println("-------------converted ------------------");
		Flux<String> converted = Flux.range(1, 10).map(i -> {
			try {
				return convert(i);
			} catch (IOException e) {
				e.printStackTrace();
				throw Exceptions.propagate(e);
			}
		});
//		converted.subscribe(i -> System.out.println(i));
		converted.subscribe(
			    v -> System.out.println("RECEIVED: " + v),
			    e -> {
			        if (Exceptions.unwrap(e) instanceof IOException) {
			            System.out.println("Something bad happened with I/O");
			        } else {
			            System.out.println("Something bad happened");
			        }
			    }
			);
	}

	public static String convert(int i) throws IOException {
		if (i > 3) {
			throw new IOException("boom " + i);
		}
		return "OK " + i;
	}
}
